package com.ls.fw.config.zookeeper;

import static com.google.common.base.Charsets.UTF_8;
import static java.lang.Integer.parseInt;
import static org.apache.curator.utils.ZKPaths.makePath;
import static org.apache.zookeeper.CreateMode.PERSISTENT;
import static org.slf4j.LoggerFactory.getLogger;

import java.nio.charset.Charset;
import java.util.ArrayDeque;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;

import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.transaction.CuratorTransactionFinal;
import org.apache.curator.framework.api.transaction.TransactionOp;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.zookeeper.KeeperException;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.ls.fw.commons.core.utils.collections.MapUtils;
import com.ls.fw.config.zookeeper.config.ZookeeperPropertySourceLocator;
import com.ls.fw.config.zookeeper.listener.ZookeeperNodeListener;
import com.ls.fw.config.zookeeper.op.CreateNodeOp;
import com.ls.fw.config.zookeeper.op.DeleteNodeOp;
import com.ls.fw.config.zookeeper.op.ZookeeperOp;
import com.ls.fw.config.zookeeper.op.ZookeeperTransaction;

/**
 * @author Shulnyaev Roman
 */
public class CuratorZookeeperClient implements ZookeeperClient {
    private CuratorFramework client;
    private final Map<String, TreeCache> subtreeCaches = new HashMap<>();
    private final Map<String, PathChildrenCache> nodeChildrenCaches = new HashMap<>();

    private static final Logger logger = getLogger(CuratorZookeeperClient.class);
	
    private List<String> contexts;
    
    private ZookeeperNodeListener zookeeperNodeListener;
    
    private String instanceRoot;
    
    
    public CuratorZookeeperClient(CuratorFramework client) {
		super();
		this.client = client;
	}
    public CuratorZookeeperClient(ZookeeperPropertySourceLocator locator,CuratorFramework client) {
		super();
		this.client = client;
		this.contexts = locator.getContexts();
		this.instanceRoot = locator.getInstanceRoot();
	}
    public CuratorZookeeperClient(List<String> contexts,String instanceRoot,CuratorFramework client) {
		super();
		this.contexts = contexts;
		this.client = client;
		this.instanceRoot = instanceRoot;
	}

    public List<String> getContexts() {
		return contexts;
	}
	public void setContexts(List<String> contexts) {
		this.contexts = contexts;
	}
	public ZookeeperNodeListener getZookeeperNodeListener() {
		return zookeeperNodeListener;
	}
	public void setZookeeperNodeListener(ZookeeperNodeListener zookeeperNodeListener) {
		this.zookeeperNodeListener = zookeeperNodeListener;
	}
	public CuratorFramework getClient() {
		return client;
	}
    public void init() throws Exception {
    	if(zookeeperNodeListener != null){
    		if(this.contexts!=null && !this.contexts.isEmpty()){
    			for (String string : contexts) {
    				if (!string.startsWith("/")) {
    					string = "/" + string;
    				}
    				this.addSubtreeListener(zookeeperNodeListener,string);
    			}
    		}
    	}
    }
    
    public void destroy() throws Exception {
        closeSubtreeCaches();
        closeNodeChildrenCaches();
        closeClient();
    }

    private void closeClient() {
        client.close();
    }

    @Override
    public String keyToPath(String nodeKey){
		return getRealPath(nodeKey);
	}

    @Override
    public String pathToKey(String nodePath){
		String root = getRoot();
		return pathToKey(root, nodePath);
	}

    @Override
    public String pathToKey(String _root,String nodePath){
    	String root = getValidPath(_root);
		if (nodePath==null || nodePath.length() <= root.length() || !nodePath.startsWith(root)) {
			return null;
		}
		return nodePath.substring(root.length()+1, nodePath.length());
	}
    @Override
    public String getDataByNodeKey(String key){
    	String path = keyToPath(key);
    	if(this.nodeExists(path)){
    		return this.getString(path);
    	}
    	return null;
    }

    @Override
    public void setDataByNodeKey(String key,String data){
    	String path = keyToPath(key);
    	this.updateNodeData(path, data);
    }

    @Override
    public void synchDataByNodeKey(Map<String,String> map) throws Exception{
    	MapUtils.traversal(map, new MapUtils.KeyValueMapper<String,String>() {

			@Override
			public boolean next(String arg0, String arg1, int arg2) throws Exception {
				String path = keyToPath(arg0);
				if(!nodeExists(path)){
					createNode(path, arg1);
				}else{
					updateNodeData(path, arg1);
				}
				return true;
			}
		});
    }
    @Override
    public String getRoot(){
    	return instanceRoot;
    }
	
    @Override
    public List<String> getChildren(String _path) throws Exception{
    	return this.client.getChildren().forPath(getValidPath(_path));
    }

    @Override
    public void setNamespace(String newNamespace) throws Exception{
    	client = client.usingNamespace(newNamespace);
    }
    
    @Override
    public void updateServerList(String connectString) throws Exception{
    	client.getZookeeperClient()
		.getZooKeeper().updateServerList(connectString);
    }
    
    // ****************************** //

    public void newTreeCache(String context,TreeCacheListener listener){
    	if (!context.startsWith("/")) {
			context = "/" + context;
		}
		try {
			if(this.client.checkExists().forPath(context)==null){
				client.create().creatingParentsIfNeeded().forPath(context); 
			}
		} catch (Exception e1) {
			e1.printStackTrace();
		}
		try {
			
			TreeCache cache = TreeCache.newBuilder(this.client, context).build();
			cache.getListenable().addListener(listener);
			cache.start();
			subtreeCaches.put(context, cache);
			// no race condition since ZookeeperAutoConfiguration.curatorFramework
			// calls curator.blockUntilConnected
		} catch (KeeperException.NoNodeException e) {
			// no node, ignore
		} catch (Exception e) {
			logger.error("Error initializing listener for context " + context, e);
		}
    }
    
    @Override
    public  boolean nodeExists(String _path) {
    	String path = getValidPath(_path);
        try {
            return client.checkExists().forPath(path) != null;
        }
        catch (Exception exception) {
            throw new ZookeeperException(exception);
        }
    }

    // ****************************** //

    @Override
    public  void createNode(String path) {
        createNode(path, "");
    }

    @Override
    public  void createNode(String _path, String data) {
    	String path = getValidPath(_path);
        try {
        	if(data==null){
        		client.create().creatingParentsIfNeeded().withMode(PERSISTENT).forPath(path, "".getBytes());
        	}else{
        		client.create().creatingParentsIfNeeded().withMode(PERSISTENT).forPath(path, data.getBytes());
        	}
        }
        catch (Exception exception) {
            throw new ZookeeperException(exception);
        }
    }
    protected String getValidPath(String context){
    	if (!context.startsWith("/")) {
			context = "/" + context;
		}
    	return context;
    }
    @Override
    public String getRealPath(String context){
    	if (!context.startsWith("/")) {
			context = "/" + context;
		}
    	String root = getRoot();
    	if(StringUtils.isNotBlank(root)){
    		if (!root.startsWith("/")) {
    			root = "/" + root;
    		}
    		context = root + context;
    	}
    	return context;
    }
    @Override
    public  void ensureNode(String _path) {
    	String path = getValidPath(_path);
        try {
            if (!nodeExists(path)) {
                createNode(path);
            }
        }
        catch (ZookeeperException e) {
            if (!(e.getCause() instanceof KeeperException.NodeExistsException)) {
                throw e;
            }
            logger.error(e.getMessage(),e);
        }
    }
    @Override
    public  void ensureNode(String _path,String data) {
    	String path = getValidPath(_path);
        try {
            if (!nodeExists(path)) {
                createNode(path,data);
            }else{
            	this.updateNodeData(path, data);
            }
        }
        catch (ZookeeperException e) {
            if (!(e.getCause() instanceof KeeperException.NodeExistsException)) {
                throw e;
            }
            logger.error(e.getMessage(),e);
        }
    }


    // ****************************** //

    @Override
    public  void updateNodeData(String _path, String data) {
    	String path = getValidPath(_path);
        try {
        	if(data!=null){
        		client.setData().forPath(path, data.getBytes());
        	}else{
        		client.setData().forPath(path, "".getBytes());
        	}
        }
        catch (Exception exception) {
            throw new ZookeeperException(exception);
        }
    }
     
    @Override
    public  void deleteNode(String _path) {
    	this.deleteNode(_path, true);
    }
    @Override
    public  void deleteNode(String _path,boolean check) {
    	String path = getValidPath(_path);
        try {
        	if(check && this.nodeExists(path)){
        		client.delete().deletingChildrenIfNeeded().forPath(path);
        	}else if(!check){
        		client.delete().deletingChildrenIfNeeded().forPath(path);
        	}
        }
        catch (Exception exception) {
            throw new ZookeeperException(exception);
        }
    }

    // ****************************** //

    @Override
    public  void executeTransaction(ZookeeperTransaction transaction) {
        if (transaction.isEmpty()) {
            return;
        }

        try {
        	TransactionOp curatorTransaction = client.transactionOp();
            for (ZookeeperOp op : transaction) {
                String _path = op.getNodePath();
            	String path = getValidPath(_path);

                if (op instanceof CreateNodeOp) {
                    byte[] data = ((CreateNodeOp)op).getNodeData().getBytes(UTF_8);
                    curatorTransaction = (TransactionOp) curatorTransaction.create().forPath(path, data).get();
                }
                else if (op instanceof DeleteNodeOp) {
                    curatorTransaction = (TransactionOp) curatorTransaction.delete().forPath(path).get();
                }
            }
            ((CuratorTransactionFinal)curatorTransaction).commit();
        }
        catch (Exception exception) {
            throw new ZookeeperException(exception);
        }
    }

    // ****************************** //

    @Override
    public  String getString(String path) {
        return new String(getData(path));
    }

    private byte[] getData(String _path) {
    	String path = getValidPath(_path);
        try {
            return client.getData().forPath(path);
        }
        catch (Exception exception) {
            throw new ZookeeperException(exception);
        }
    }

    @Override
    public  int getInt(String path) {
        return parseInt(getString(path));
    }


	public String getEventDesc(TreeCacheEvent event) {
		StringBuilder out = new StringBuilder();
		out.append("type=").append(event.getType());
		if(event.getData()!=null){
			out.append(", path=").append(event.getData().getPath());
			byte[] data = event.getData().getData();
			if (data != null && data.length > 0) {
				out.append(", data=").append(new String(data, Charset.forName("UTF-8")));
			}
		}
		return out.toString();
	}
	public String getEventDesc(PathChildrenCacheEvent event) {
		StringBuilder out = new StringBuilder();
		out.append("type=").append(event.getType());
		if(event.getData()!=null){
			out.append(", path=").append(event.getData().getPath());
			byte[] data = event.getData().getData();
			if (data != null && data.length > 0) {
				out.append(", data=").append(new String(data, Charset.forName("UTF-8")));
			}
		}
		return out.toString();
	}
    // ****************************** //

    @Override
    public  void addSubtreeListener(String _subtreeRootNodePath, final ZookeeperNodeListener listener) {
    	String subtreeRootNodePath = getValidPath(_subtreeRootNodePath);
    	addSubtreeListener(listener, subtreeRootNodePath);
    }
	public void addSubtreeListener(final ZookeeperNodeListener listener, String subtreeRootNodePath) {
		ensureNode(subtreeRootNodePath);
        if(listener==null){
        	return ;
        }
        //监听到指定节点下所有节点的变化
        TreeCache subtreeCache =
            subtreeCaches.containsKey(subtreeRootNodePath) ?
            subtreeCaches.get(subtreeRootNodePath) :
            	TreeCache.newBuilder(this.client, subtreeRootNodePath).build();

        subtreeCache.getListenable().addListener(
            (curatorFramework, event) -> {
                 TreeCacheEvent.Type eventType = event.getType();
                 ChildData descendant = event.getData();
                 if(eventType!=null){
                	 switch (eventType) {
                	 case NODE_ADDED:
                		 DateTime creationTime = new DateTime(descendant.getStat().getCtime());
                		 listener.onZookeeperNodeAdded(listener,event,getEventDesc(event),
                				 descendant.getPath(),
                				 pathToKey(descendant.getPath()),
                				 getData(descendant),
                				 creationTime);
                		 break;
                	 case NODE_REMOVED:
                		 listener.onZookeeperNodeDeleted(listener,event,getEventDesc(event),descendant.getPath(),
                				 pathToKey(descendant.getPath()));
                		 break;
                	 case NODE_UPDATED:
                		 DateTime modificationTime = new DateTime(descendant.getStat().getMtime());
                		 listener.onZookeeperNodeUpdated(listener,event,getEventDesc(event),
                				 descendant.getPath(),
                				 pathToKey(descendant.getPath()),
                				 getData(descendant),
                				 modificationTime);
                		 break;
					  default:
						break;
                	 }
                 }
            });

        if (subtreeCaches.containsKey(subtreeRootNodePath)) {
            ChildData subtreeRootNode = subtreeCache.getCurrentData(subtreeRootNodePath);

            if (subtreeRootNode != null) {
            	TreeCacheEvent event = new TreeCacheEvent(TreeCacheEvent.Type.NODE_ADDED,subtreeRootNode);
                listener.onZookeeperNodeAdded(listener,event,getEventDesc(event),
                    subtreeRootNodePath,
   				    pathToKey(subtreeRootNodePath),
                    getData(subtreeRootNode),
                    new DateTime(subtreeRootNode.getStat().getCtime()));
            }

            Queue<String> subtreeTraverseQueue = new ArrayDeque<>();
            subtreeTraverseQueue.add(subtreeRootNodePath);

            while (!subtreeTraverseQueue.isEmpty()) {
                String currentSubtreeNodePath = subtreeTraverseQueue.remove();
                Map<String, ChildData> currentSubtreeNodeChildren = subtreeCache.getCurrentChildren(currentSubtreeNodePath);

                if (currentSubtreeNodeChildren != null) {
                    for (Map.Entry<String, ChildData> currentSubtreeNodeChild : currentSubtreeNodeChildren.entrySet()) {
                        String currentSubtreeNodeChildPath = makePath(currentSubtreeNodePath, currentSubtreeNodeChild.getKey());
                        String currentSubtreeNodeChildData = getData(currentSubtreeNodeChild.getValue());
                        DateTime currentSubtreeNodeChildCreationTime = new DateTime(currentSubtreeNodeChild.getValue().getStat().getCtime());
                        TreeCacheEvent event = new TreeCacheEvent(TreeCacheEvent.Type.NODE_ADDED,currentSubtreeNodeChild.getValue());
                        listener.onZookeeperNodeAdded(listener,event,getEventDesc(event),
                            currentSubtreeNodeChildPath,
           				    pathToKey(subtreeRootNodePath),
                            currentSubtreeNodeChildData,
                            currentSubtreeNodeChildCreationTime);

                        subtreeTraverseQueue.add(currentSubtreeNodeChildPath);
                    }
                }
            }
        }
        else {
            subtreeCaches.put(subtreeRootNodePath, subtreeCache);

            try {
                subtreeCache.start();
            }
            catch (Exception exception) {
                throw new ZookeeperException(exception);
            }
        }
	}

    private void closeSubtreeCaches() {
        for (TreeCache treeCache : subtreeCaches.values()) {
            try {
                treeCache.close();
            }
            catch (Exception e) {
            	logger.error(e.getMessage(),e);
            }
        }

        subtreeCaches.clear();
    }

    // ****************************** //

    @Override
    public  List<String> getNodeChildNames(String _path) {
    	String path = getValidPath(_path);
        try {
            return client.getChildren().forPath(path);
        }
        catch (Exception exception) {
            throw new ZookeeperException(exception);
        }
    }

    @Override
    public  void addNodeChildrenListener(String _path, final ZookeeperNodeListener listener) {
    	String path = getValidPath(_path);
    	addNodeChildrenListener(listener, path);
    }
	public void addNodeChildrenListener(final ZookeeperNodeListener listener, String path) {
		ensureNode(path);
        if(listener==null){
        	return ;
        }
        //只能监听指定节点下一级节点的变化
        PathChildrenCache nodeChildrenCache =
            nodeChildrenCaches.containsKey(path) ?
            nodeChildrenCaches.get(path) :
            new PathChildrenCache(client, path, true);

        nodeChildrenCache.getListenable().addListener(
            (curatorFramework, event) -> {
                PathChildrenCacheEvent.Type eventType = event.getType();
                ChildData child = event.getData();
                if(eventType!=null){
	               	 switch (eventType) {
						case CHILD_ADDED:
							  DateTime creationTime = new DateTime(child.getStat().getCtime());
			                    listener.onZookeeperNodeAdded(listener,event,getEventDesc(event),
			                        child.getPath(),
			       				    pathToKey(child.getPath()),
			                        getData(child),
			                        creationTime);
							break;
						case CHILD_REMOVED:
						       listener.onZookeeperNodeDeleted(listener,event,getEventDesc(event),child.getPath(),
				       				    pathToKey(child.getPath()));
					           break;
						case CHILD_UPDATED:
							  DateTime modificationTime = new DateTime(child.getStat().getMtime());
			                    listener.onZookeeperNodeUpdated(listener,event,getEventDesc(event),
			                        child.getPath(),
			       				    pathToKey(child.getPath()),
			                        getData(child),
			                        modificationTime);
							break;
						default:
							break;
	               	 }
                }
            });

        if (!nodeChildrenCaches.containsKey(path)) {
            nodeChildrenCaches.put(path, nodeChildrenCache);

            try {
                nodeChildrenCache.start();
            }
            catch (Exception exception) {
                throw new ZookeeperException(exception);
            }
        }
        else {
            for (ChildData child : nodeChildrenCache.getCurrentData()) {
                DateTime creationTime = new DateTime(child.getStat().getCtime());
                TreeCacheEvent event = new TreeCacheEvent(TreeCacheEvent.Type.NODE_ADDED,child);
                listener.onZookeeperNodeAdded(listener,event,getEventDesc(event),
                    child.getPath(),
   				    pathToKey(child.getPath()),
                    getData(child),
                    creationTime);
            }
        }
	}

    private String getData(ChildData child){
    	String data = null;
        if(child!=null && child.getData()!=null){
        	data = new String(child.getData());
        }
        return data;
    }
    
    private void closeNodeChildrenCaches() {
        for (PathChildrenCache nodeChildrenCache : nodeChildrenCaches.values()) {
            try {
                nodeChildrenCache.close();
            }
            catch (Exception e) {
            	logger.error(e.getMessage(),e);
            }
        }

        nodeChildrenCaches.clear();
    }
	public String getInstanceRoot() {
		return instanceRoot;
	}
	public void setInstanceRoot(String instanceRoot) {
		this.instanceRoot = instanceRoot;
	}

    // ****************************** //

 }
